package com.lhd.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.lhd.bean.KeywordBean;
import com.lhd.bean.KeywordStats;
import com.lhd.utils.DateFormatUtil;
import com.lhd.utils.MyClickHouseUtil;
import com.lhd.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;

public class KeyWordTop10App {
    public static void main(String[] args) throws Exception {
        // todo 流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取 ods_traffic 原始数据
        String topic = "dwd_traffic_page_log" ;
        String groupId = "xxx" ;
        DataStreamSource<String> kfkSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        // todo 打印数据 查看格式
//        kfkSource.print();

        // todo 对数据进行判断 是否为关键词搜索数据
        SingleOutputStreamOperator<JSONObject> searchDs = kfkSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject json = JSONObject.parseObject(value);
                    // todo 错误数据抛异常
                    if (json != null) {
                        if ("search".equals(json.getJSONObject("page").getString("last_page_id")) && "keyword".equals(json.getJSONObject("page").getString("item_type"))) {
                            out.collect(json);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }


            }
        });

//        searchDs.print("关键词搜索的数据--->");

        // todo 对数据设置水位线
        SingleOutputStreamOperator<JSONObject> searchWms = searchDs.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(1L))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                }));

//        searchWms.print("关键词搜索的数据--->");


        SingleOutputStreamOperator<KeywordBean> aggregate = searchWms.
                keyBy(s -> s.getJSONObject("page").getString("item"))
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .aggregate(new AggregateFunction<JSONObject, Tuple5<String, String, String, Long, Long>, Tuple5<String, String, String, Long, Long>>() {
                    @Override
                    public Tuple5<String, String, String, Long, Long> createAccumulator() {
                        return Tuple5.of("", "", "", 0L, 0L);
                    }

                    @Override
                    public Tuple5<String, String, String, Long, Long> add(JSONObject value, Tuple5<String, String, String, Long, Long> accumulator) {
                        accumulator.f0 = value.getJSONObject("page").getString("item");
                        accumulator.f1 = value.getJSONObject("page").getString("last_page_id");
                        accumulator.f2 = value.getJSONObject("common").getString("uid");
                        accumulator.f3 += 1L;
                        accumulator.f4 += 1L;
                        return accumulator;
                    }

                    @Override
                    public Tuple5<String, String, String, Long, Long> getResult(Tuple5<String, String, String, Long, Long> accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Tuple5<String, String, String, Long, Long> merge(Tuple5<String, String, String, Long, Long> a, Tuple5<String, String, String, Long, Long> b) {
                        return null;
                    }
                }, new ProcessWindowFunction<Tuple5<String, String, String, Long, Long>, KeywordBean, String, TimeWindow>() {
                    ValueState<Set<String>> uvState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<Set<String>> descriptor =
                                new ValueStateDescriptor<>("uvState", new TypeHint<Set<String>>() {
                                }.getTypeInfo());
                        uvState = getRuntimeContext().getState(descriptor);
                    }

                    @Override
                    public void process(String s, ProcessWindowFunction<Tuple5<String, String, String, Long, Long>, KeywordBean,
                            String, TimeWindow>.Context context,
                                        Iterable<Tuple5<String, String, String, Long, Long>> elements,
                                        Collector<KeywordBean> out) throws Exception {
                        // 初始化或获取当前状态
                        Set<String> visitedUsers = uvState.value();
                        if (visitedUsers == null) {
                            visitedUsers = new HashSet<>();
                        }

                        long viewCount = 0L;
                        String source = "";

                        // 遍历窗口内的所有元素
                        for (Tuple5<String, String, String, Long, Long> element : elements) {
                            viewCount = element.f3;

                            // 获取用户ID
                            String userId = "";
                            if (element.f2 != null && !"".equals(element.f2)) {
                                userId = element.f2;
                            }

                            // 获取来源类型
                            if (element.f1 != null && !"".equals(element.f1)) {
                                source = element.f1;
                            }

                            // 如果用户未访问过该关键词，则添加到集合中
                            if (!userId.isEmpty()) {
                                visitedUsers.add(userId);
                            }
                        }

                        // 创建结果对象
                        KeywordBean keywordBean = new KeywordBean(
                                new Timestamp(context.window().getStart()),
                                new Timestamp(context.window().getEnd()),
                                s, // keyword
                                source,
                                viewCount,
                                (long) visitedUsers.size() // uv count
                        );

                        // 更新状态
                        uvState.update(visitedUsers);

                        // 输出结果
                        out.collect(keywordBean);
                    }
                });


        aggregate.print("聚合结果----->");


        // todo 读取业务数据
        String topic_name = "topic_db";
        String groupId_name = "keyword_top10_app_group_name" ;

        DataStreamSource<String> dbSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic_name, groupId_name).setStartFromEarliest());

//        dbSource.print("业务数据----->");

        // todo 创建测流
        OutputTag<JSONObject> order_info = new OutputTag<JSONObject>("order_info"){};
        OutputTag<JSONObject> order_detail = new OutputTag<JSONObject>("order_detail"){};
        OutputTag<JSONObject> payment_info = new OutputTag<JSONObject>("payment_info"){};


        // todo 业务数据取 gmall 下的 订单表和订单明细表
        SingleOutputStreamOperator<JSONObject> process = dbSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                JSONObject dbJson = JSONObject.parseObject(value);
                if (dbJson != null) {
                    if ("gmall".equals(dbJson.getString("database")) && "order_info".equals(dbJson.getString("table"))) {
                        ctx.output(order_info, dbJson);
                    } else if ("gmall".equals(dbJson.getString("database")) && "order_detail".equals(dbJson.getString("table"))) {
                        ctx.output(order_detail, dbJson);
                    }else if("gmall".equals(dbJson.getString("database")) && "payment_info".equals(dbJson.getString("table"))){
                        ctx.output(payment_info, dbJson);
                    }
                }
                out.collect(dbJson);
            }
        });

//        process.getSideOutput(order_info).print("order_info---->");
//        process.getSideOutput(order_detail).print("order_detail---->");

        // todo 对 订单表 和 订单明细表 分别设置水位线
        SingleOutputStreamOperator<JSONObject> orderInfoWms = process.getSideOutput(order_info).map(s->s.getJSONObject("data")).
                assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2L)))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return DateFormatUtil.toTs(element.getString("create_time"),true);
                            }
                        }));

        SingleOutputStreamOperator<JSONObject> orderDetailWms = process.getSideOutput(order_detail).map(s->s.getJSONObject("data")).
                assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return DateFormatUtil.toTs(element.getString("create_time"),true);
                            }
                        })
                );

        SingleOutputStreamOperator<JSONObject> paymentInfoWms = process.getSideOutput(payment_info).map(s->s.getJSONObject("data")).
                assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return DateFormatUtil.toTs(element.getString("create_time"),true);
                            }
                        })
                );

//        orderInfoWms.print("订单表数据--->");
//        orderDetailWms.print("订单明细表数据--->");
//        paymentInfoWms.print("支付表--->");

        SingleOutputStreamOperator<JSONObject> orderInfoDetailWide = orderInfoWms.keyBy(s -> s.getString("id"))
                .intervalJoin(orderDetailWms.keyBy(s ->s.getString("order_id")))
                .between(Time.seconds(-10), Time.seconds(10L))
                .process(new ProcessJoinFunction<JSONObject, JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject left, JSONObject right, ProcessJoinFunction<JSONObject, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                        for (String s : right.keySet()) {
                            left.put("order_detail_"+s,right.get(s));
                        }

                        out.collect(left);


                    }
                });

//        orderInfoDetailWide.print("订单表和订单明细表宽表--->");

        SingleOutputStreamOperator<JSONObject> orderInfoDetailPaymentWide = orderInfoDetailWide.keyBy(s -> s.getString("id"))
                .intervalJoin(paymentInfoWms.keyBy(s -> s.getString("order_id")))
                .between(Time.seconds(-10), Time.seconds(10L))
                .process(new ProcessJoinFunction<JSONObject, JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject left, JSONObject right, ProcessJoinFunction<JSONObject, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        for (String s : right.keySet()) {
                            left.put("payment_" + s, right.get(s));
                        }
                        out.collect(left);
                    }
                });

        orderInfoDetailPaymentWide.print("订单表和订单明细表和支付表宽表--->");

//        searchWms.map(s->s.getJSONObject("common").getString("uid").equals("100")).print("uid--->");
//        orderInfoDetailPaymentWide.map(s -> s.getString("user_id").equals("100")).print("user_id--->");

        SingleOutputStreamOperator<JSONObject> process1 = searchWms.keyBy(s -> s.getJSONObject("common").getString("uid"))
                .intervalJoin(orderInfoDetailPaymentWide.keyBy(s -> s.getString("user_id")))
                .between(Time.seconds(-10), Time.seconds(10))
                .process(new ProcessJoinFunction<JSONObject, JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject left, JSONObject right, ProcessJoinFunction<JSONObject, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        right.put("page_keyword", left.getJSONObject("page").getString("item"));
                        right.put("page_keyword", left.getJSONObject("page").getString("item_type"));
                        right.put("page_source", left.getJSONObject("page").getString("source"));
                        out.collect(right);
                    }
                });

        process1.print("最终结果--->");



        // 将 KeywordBean 转换为 KeywordStats 并写入 ClickHouse
        SingleOutputStreamOperator<KeywordStats> keywordStatsDs = aggregate.map(keywordBean -> {
            KeywordStats keywordStats = new KeywordStats();
            keywordStats.setStatsTime(keywordBean.getStt().toString());
            keywordStats.setStatsType("10s"); // 10秒窗口
            keywordStats.setKeyword(keywordBean.getKeyword());
            keywordStats.setSearchCount(keywordBean.getViewCount());
            keywordStats.setUvCount(keywordBean.getUvCount());
            keywordStats.setPvCount(keywordBean.getViewCount());
            keywordStats.setCartAddCount(0L); // 暂时设置为0，实际应用中需要根据业务数据计算
            keywordStats.setPaymentAmount(0.0); // 暂时设置为0，实际应用中需要根据业务数据计算
            keywordStats.setRank(0); // 暂时设置为0，后续可以排序
            keywordStats.setCreateTime(new Date());
            return keywordStats;
        });

        // 创建ClickHouse Sink
        String sql = "INSERT INTO keyword_stats VALUES(?,?,?,?,?,?,?,?,?,?)";
        SinkFunction<KeywordStats> sinkFunction = MyClickHouseUtil.getSinkFunction(sql);
        keywordStatsDs.addSink(sinkFunction);


        // todo 执行
        env.execute();
    }
}
